package com.atguigu.Func;

import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.Util.DruidDSUtil;

import com.atguigu.Util.JedisUtil;
import com.atguigu.common.GmallConfig;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

import java.sql.PreparedStatement;
import java.util.Collection;
import java.util.Map;
import java.util.Set;

/**
 * @author hjy
 * @create 2023/3/10 10:21
 */
//{"database":"gmall-220623-flink","table":"comment_info","type":"insert","ts":1669162958,"xid":1111,"xoffset":13941,"data":{"id":1595211185799847960,"user_id":119,"nick_name":null,"head_img":null,"sku_id":31,"spu_id":10,"order_id":987,"appraise":"1204","comment_txt":"评论内容：48384811984748167197482849234338563286217912223261","create_time":"2022-08-02 08:22:38","operate_time":null}}
public class DimJdbcSink extends RichSinkFunction<JSONObject> {
    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        //双写
        JSONObject data = value.getJSONObject("data");
        if ("update".equals(value.getString("type"))){
            Jedis jedis = JedisUtil.getJedis();
            String redisKey="DIM:"+value.getString("sink_table").toUpperCase()+":"+ data.getString("id");
            //这里要将data里的key都换成大写  所以要处理一下
            JSONObject jsonObject = new JSONObject();
            Set<Map.Entry<String, Object>> entries = data.entrySet();
            for (Map.Entry<String, Object> entry : entries) {
                String key = entry.getKey().toUpperCase();
                Object entryValue = entry.getValue();
                if (entryValue==null||"".equals(entryValue)){
                    jsonObject.put(key,"");
                }else {
                    jsonObject.put(key,value.toString());
                }
            }
            jedis.setex(redisKey,24*3600,jsonObject.toJSONString());
            jedis.close();
        }


        //建立连接
        DruidPooledConnection phoenixConn = DruidDSUtil.getPhoenixConn();
        //拼接sql
        String sql= genUpsertSQL(value.getString("sink_table"), data);
        System.out.println("upsert语句："+sql);
        //输出数据
        PreparedStatement preparedStatement = phoenixConn.prepareStatement(sql);
        preparedStatement.execute();
        //Phoenix中的DML语句要手动commit一下 或者将setAutoCommit()设置为true
        //mysql中这个AutoCommit默认为true，但是phoenix中默认是false

        phoenixConn.commit();
        //关闭连接
        phoenixConn.close();
    }
    //upsert into db.t (id,name,url) values('1001','zs','./atguigu')
    //拼接的sql upsert into db.tn(id,name,sex) values('1001','zs','male)
    private String genUpsertSQL(String sink_table, JSONObject data) {
        Set<String> columns = data.keySet();
        Collection<Object> values = data.values();
        return "upsert into "
                + GmallConfig.HBASE_SCHEMA
                +"."+sink_table
                +"("
                + StringUtils.join(columns,",")
                +") values ('"
                +StringUtils.join(values,"','")
                +"')";
    }
}
